// -*-mode:c++; coding:utf-8-*-
#ifndef _XBASE_SSTABLE_BLOCK_CACHE_MANAGER_
#define _XBASE_SSTABLE_BLOCK_CACHE_MANAGER_

#include <string>
#include <cassert>

#include <boost/noncopyable.hpp>
#include <boost/pool/detail/singleton.hpp>
#include <boost/pool/pool_alloc.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/multi_index_container.hpp>
#include <boost/multi_index/tag.hpp>
#include <boost/multi_index/indexed_by.hpp>
#include <boost/multi_index/hashed_index.hpp>
#include <boost/multi_index/ordered_index.hpp>
#include <boost/multi_index/mem_fun.hpp>

#include <config.hpp>

#include "sstable_block_cache_base.hpp"

namespace xbase 
{

	namespace details
	{
		class sstable_block_cache_manager_impl_fast
			: private boost::noncopyable
		{
		private:
			typedef boost::fast_pool_allocator<sstable_block_cache_manager_impl_fast> allocator_type;

			static
			sstable_block_cache_manager_impl_fast *s_instance;

			static
			allocator_type s_allocator;

		public:
			static inline
			sstable_block_cache_manager_impl_fast &instance() {
				return *s_instance;
			}

			static inline
			void init_instance(uint64_t max_memory) {
				assert(s_instance == NULL);
				s_instance = s_allocator.allocate();
				::new(s_instance) sstable_block_cache_manager_impl_fast(max_memory);
			}

			static inline
			void reset_instance() {
				s_instance->~sstable_block_cache_manager_impl_fast();
				s_allocator.deallocate(s_instance);
				s_instance = NULL;
			}

		public:
			typedef cache_key cache_key_type;
			typedef cache_node cache_node_type;

			typedef cache_key_type::file_id_type file_id_type;

			typedef default_char_allocator data_allocator_type;
			typedef boost::fast_pool_allocator<cache_node_type> cache_node_allocator_type;

			struct cache_node_wrapper
			{
				explicit
				cache_node_wrapper(cache_node_type *node)
					: m_node(node) {}

				cache_node_type *m_node;

				std::size_t ref_count() const {
					return m_node->m_value.ref_count();
				}

				void reset_ref_count(std::size_t value) const {
					return m_node->m_value.reset_ref_count(value);
				}

				cache_key_type key() const {
					return m_node->m_key;
				}

#ifndef NDEBUG
				file_id_type file_id() const {
					return m_node->m_key.m_file_id;
				}
#endif	// NDEBUG

				void inc_ref_count() {
					m_node->m_value.inc_ref_count();
				}
			
				void dec_ref_count() {
					m_node->m_value.dec_ref_count();
				}
			};

			struct key_hasher
			{
				std::size_t operator()(const cache_key_type &key) const {
					return key.hash();
				}
			};

			struct ref_count {};
			struct key {};
#ifndef NDEBUG
			struct file_id {};
#endif	// NDEBUG
		
			typedef boost::multi_index::multi_index_container<
				cache_node_wrapper,
				boost::multi_index::indexed_by<
					boost::multi_index::ordered_non_unique<
						boost::multi_index::tag<ref_count>,
						BOOST_MULTI_INDEX_CONST_MEM_FUN(cache_node_wrapper, std::size_t, ref_count)
						>,
							boost::multi_index::hashed_unique<
								boost::multi_index::tag<key>,
								BOOST_MULTI_INDEX_CONST_MEM_FUN(cache_node_wrapper, cache_key_type, key),
								key_hasher
								>
#ifndef NDEBUG
									,
									boost::multi_index::ordered_non_unique<
										boost::multi_index::tag<file_id>,
										BOOST_MULTI_INDEX_CONST_MEM_FUN(cache_node_wrapper, file_id_type, file_id)
										>
#endif	// NDEBUG
					>
				> block_cache;

			typedef boost::multi_index::index<block_cache, ref_count>::type ref_count_index_type;
			typedef boost::multi_index::index<block_cache, key>::type key_index_type;
#ifndef NDEBUG
			typedef boost::multi_index::index<block_cache, file_id>::type file_id_index_type;
#endif	// NDEBUG
		
		public:
			inline
			cache_node_type *allocate(std::size_t block_size) {
				cache_node_type *node = m_cache_node_allocator.allocate();
				::new(node) cache_node_type();
				node->m_value.m_block_data = m_data_allocator.allocate(block_size);
				node->m_value.m_alloc_size = block_size;
				return node;
			}

			inline
			void deallocate(cache_node_type *node) {
				node->~cache_node_type();
				m_data_allocator.deallocate(node->m_value.m_block_data,
							    node->m_value.m_alloc_size);
				m_cache_node_allocator.deallocate(node);
			}

			inline
			bool commit(cache_node_type *node) {
				return commit_with_ref_count(node, 0);
			}

			inline
			bool commit_and_checkout(cache_node_type *node) {
				return commit_with_ref_count(node, 1);
			}
		
			cache_node_type *checkout(const cache_key_type &key);

			bool checkin(cache_node_type *node);

			sstable_block_cache_manager_impl_fast(uint64_t max_memory)
				: m_max_memory(max_memory)
				, m_free_memory(max_memory)
				, m_busy_memory(0)
				, m_max_busy_memory(0)
				, m_block_number(0)
				, m_checkout_counter(0)
				, m_checkout_counter_ok(0) {}

			~sstable_block_cache_manager_impl_fast();

#ifndef NDEBUG
			void clear_sstable_cache(file_id_type file_id);
#endif	// NDEBUG

			// 还有多少缓存可用，包含缓存但未被使用的块
			uint64_t available_memory() const {
				return m_max_memory - m_busy_memory;
			}

			std::size_t cached_block_number() const {
				return m_block_number;
			}

			uint64_t cached_block_size() const {
				return m_max_memory - m_free_memory;
			}

			double hit_rate() const {
				return (double)m_checkout_counter_ok / (double)m_checkout_counter;
			}

			std::string to_string() const;

			inline
			bool is_cached(const cache_key_type &cache_key) {
				boost::mutex::scoped_lock lock(m_mutex);
				const key_index_type &key_index = m_block_cache.get<key>();
				const key_index_type::const_iterator iter = key_index.find(cache_key);
				return (iter != key_index.end());
			}

		protected:
			bool commit_with_ref_count(cache_node_type *node,
						   std::size_t ref_count);
		
		private:
			block_cache m_block_cache;
			boost::mutex m_mutex;

			const uint64_t m_max_memory; // 最大缓存大小
			uint64_t m_free_memory; // 未被占用的内存大小
			uint64_t m_busy_memory;	     // 正在被使用的块的总大小，即那些不能被淘汰的部分
			uint64_t m_max_busy_memory;  // 正被使用的块的总大小达到过的最大值，可以参考作为缓存大小的配置值
			std::size_t m_block_number;  // 缓存的块数

			std::size_t m_checkout_counter;
			std::size_t m_checkout_counter_ok;

			data_allocator_type m_data_allocator;
			cache_node_allocator_type m_cache_node_allocator;
		};

	} // namespace details

	//typedef boost::details::pool::singleton_default<details::sstable_block_cache_manager_impl_fast> sstable_block_cache_manager;
	typedef details::sstable_block_cache_manager_impl_fast sstable_block_cache_manager;

} // namespace xbase

#endif	// _XBASE_SSTABLE_BLOCK_CACHE_MANAGER_
